package com.alibaba.nacos.common.remote.client;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.exception.runtime.NacosRuntimeException;
import com.alibaba.nacos.api.remote.RequestCallBack;
import com.alibaba.nacos.api.remote.request.ClientDetectionRequest;
import com.alibaba.nacos.api.remote.request.ConnectResetRequest;
import com.alibaba.nacos.api.remote.request.HealthCheckRequest;
import com.alibaba.nacos.api.remote.request.Request;
import com.alibaba.nacos.api.remote.response.ClientDetectionResponse;
import com.alibaba.nacos.api.remote.response.ConnectResetResponse;
import com.alibaba.nacos.api.remote.response.Response;
import com.google.common.base.Stopwatch;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.Setter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.AlwaysRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;

import static com.alibaba.nacos.api.exception.NacosException.CLIENT_DISCONNECT;
import static com.alibaba.nacos.api.exception.NacosException.SERVER_ERROR;

@Slf4j
public abstract class RpcClient
        implements Cloneable {

    private final static long TIMEOUT = Long.MAX_VALUE;
    private final static int RETRY_TIMES = 1;

    protected volatile Connection currentConnection;
    protected volatile AtomicReference<RpcClientStatus> rpcClientStatus = new AtomicReference<>(RpcClientStatus.WAIT_INIT);
    private final BlockingQueue<ReconnectContext> reconnectionSignal = new ArrayBlockingQueue<>(1);
    private ServerListFactory serverListFactory;
    protected BlockingQueue<ConnectionEvent> eventLinkedBlockingQueue = new LinkedBlockingQueue<>();
    private Stopwatch lastActiveTimeStamp = Stopwatch.createUnstarted();
    protected List<ServerRequestHandler> serverRequestHandlers = new ArrayList<>();


    public void asyncRequest(Request request,
                             RequestCallBack callBack) throws NacosException {
        Throwable e = null;
        Stopwatch stopwatch = Stopwatch.createStarted();
        for (int retryTime = 0; retryTime < RETRY_TIMES && stopwatch.elapsed(TimeUnit.MILLISECONDS) < callBack.getTimeout(); retryTime++) {
            if (this.currentConnection == null || !isRunning()) {
                e = Optional.ofNullable(e).orElseGet(() -> new NacosException(CLIENT_DISCONNECT, "client not connected"));
                LockSupport.parkNanos(Math.min(100, callBack.getTimeout() / 3));
                continue;
            }

            try {
                this.currentConnection.asyncRequest(request, callBack);
                return;
            } catch (Exception exception) {
                e = exception;
            }
        }

        if (rpcClientStatus.compareAndSet(RpcClientStatus.RUNNING, RpcClientStatus.UNHEALTHY)) {
            switchServerAsync(null, true);
        }

        if (e != null) {
            throw (e instanceof NacosException nacosException) ? nacosException : new NacosException(SERVER_ERROR, e);
        } else {
            throw new NacosException(SERVER_ERROR, "fail");
        }
    }

    public RpcClient setServerListFactory(ServerListFactory serverListFactory) {
        if (this.rpcClientStatus.get() != RpcClientStatus.WAIT_INIT) {
            return this;
        }

        this.serverListFactory = serverListFactory;
        this.rpcClientStatus.compareAndSet(RpcClientStatus.WAIT_INIT, RpcClientStatus.INITIALIZED);
        return this;
    }


    RetryTemplate retryTemplate = RetryTemplate.builder().maxAttempts(RETRY_TIMES).exponentialBackoff(50, 2, 1000).retryOn(Exception.class).build();

    public final void start() {
        boolean changeState = rpcClientStatus.compareAndSet(RpcClientStatus.INITIALIZED, RpcClientStatus.STARTING);
        if (!changeState) {
            log.error("rpc client started yet");
            return;
        }

        Thread.startVirtualThread(this::handlerConnectionEvent);
        Thread.startVirtualThread(this::handleReconnectEvent);

        rpcClientStatus.set(RpcClientStatus.STARTING);

        Connection connectToServer = retryTemplate.execute(context -> {
            String serverAddress = serverListFactory.genNextServer();
            return connectToServer(ServerInfo.fromAddress(serverAddress));
        }, context -> {
            log.error("connectToServer failed", context.getLastThrowable());
            return null;
        });

        if (connectToServer != null) {
            this.currentConnection = connectToServer;
            this.rpcClientStatus.set(RpcClientStatus.RUNNING);
            this.eventLinkedBlockingQueue.offer(new ConnectionEvent(ConnectionEvent.CONNECTED, currentConnection));
        } else {
            // 当前服务器连接不上 转入异步连接其他服务器
            switchServerAsync();
        }

        this.serverRequestHandlers.add(new ConnectResetRequestHandler());
        this.serverRequestHandlers.add((request, connection) -> request instanceof ClientDetectionRequest ? new ClientDetectionResponse() : null);
    }

    private void handleReconnectEvent() {
        for (; !isShutdown(); ) {
            try {
                ReconnectContext reconnectContext = reconnectionSignal.poll(TIMEOUT, TimeUnit.MILLISECONDS);
                if (reconnectContext == null) {
                    // 未获取到数据 继续等待
                    if (lastActiveTimeStamp.elapsed(TimeUnit.MILLISECONDS) < TIMEOUT) {
                        continue;
                    }

                    // 超过connectionKeepAlive
                    if (healthCheck()) {
                        lastActiveTimeStamp.reset().start();
                        continue;
                    }

                    if (currentConnection == null) {
                        continue;
                    }

                    RpcClientStatus rpcClientStatus = RpcClient.this.rpcClientStatus.get();
                    if (rpcClientStatus == RpcClientStatus.SHUTDOWN) {
                        break;
                    }

                    boolean success = RpcClient.this.rpcClientStatus.compareAndSet(rpcClientStatus, RpcClientStatus.UNHEALTHY);
                    // 当前线程抢占不成功 继续循环
                    if (!success) {
                        continue;
                    }

                    reconnectContext = new ReconnectContext(null, false);
                }


                if (reconnectContext.serverInfo != null) {
                    final ServerInfo newServerInfo = reconnectContext.serverInfo;
                    Optional<ServerInfo> serverInfoOptional = serverListFactory.getServerList()
                            .stream()
                            .map(ServerInfo::fromAddress)
                            .filter(serverExist -> Objects.equals(serverExist.getServerIp(), newServerInfo.getServerIp()))
                            .findAny();
                    if (serverInfoOptional.isPresent()) {
                        newServerInfo.serverPort = serverInfoOptional.get().serverPort;
                    } else {
                        reconnectContext.serverInfo = null;
                    }
                }

                reconnect(reconnectContext.serverInfo, reconnectContext.onRequestFail);
            } catch (Exception e) {
                throw new UnsupportedOperationException(e);
            }
        }
    }

    private void handlerConnectionEvent() {
        for (; !isShutdown(); ) {
            try {
                ConnectionEvent take = eventLinkedBlockingQueue.take();
                if (take.isConnected()) {
                    notifyConnected(take.connection);
                } else if (take.isDisConnected()) {
                    notifyDisConnected(take.connection);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    protected void notifyConnected(Connection connection) {
    }

    protected void notifyDisConnected(Connection connection) {
    }

    protected void reconnect(ServerInfo serverInfo,
                             boolean onRequestFail) throws NacosException {
        if (onRequestFail && healthCheck()) {
            rpcClientStatus.set(RpcClientStatus.RUNNING);
            return;
        }

        RetryTemplate retryTemplate = new RetryTemplate();
        retryTemplate.setRetryPolicy(new AlwaysRetryPolicy());
        retryTemplate.setBackOffPolicy(new ExponentialBackOffPolicy());


        for (int reConnectTimes = 0; !isShutdown(); reConnectTimes++) {
            ServerInfo usingServerInfo = Optional.ofNullable(serverInfo).orElseGet(() -> ServerInfo.fromAddress(serverListFactory.genNextServer()));
            Connection newConnection = connectToServer(usingServerInfo);
            if (newConnection != null) {
                if (currentConnection != null) {
                    currentConnection.setAbandon(true);
                    closeConnection(currentConnection);
                }

                this.currentConnection = newConnection;
                rpcClientStatus.set(RpcClientStatus.RUNNING);
                eventLinkedBlockingQueue.offer(new ConnectionEvent(ConnectionEvent.CONNECTED, newConnection));
                break;
            }

            List<String> serverList = serverListFactory.getServerList();
            if (CollectionUtils.isEmpty(serverList)) {
                throw new RuntimeException("server list empty");
            }

            if (!isRunning()) {
                LockSupport.parkNanos(Math.min(30000, reConnectTimes * 10L));
            }
        }
    }

    public boolean isRunning() {
        return rpcClientStatus.get() == RpcClientStatus.RUNNING;
    }

    private void closeConnection(Connection connection) {
        if (connection == null) {
            return;
        }

        connection.close();
        eventLinkedBlockingQueue.offer(new ConnectionEvent(ConnectionEvent.DISCONNECTED, connection));
    }


    @SneakyThrows
    private boolean healthCheck() {
        if (this.currentConnection == null) {
            return false;
        }

        return retryTemplate.execute(context -> {
            try {
                Response response = this.currentConnection.request(new HealthCheckRequest(), TIMEOUT);
                return response != null && response.isSuccess();
            } catch (NacosException e) {
                throw new RuntimeException(e);
            }
        }, context -> false);
    }

    private boolean isShutdown() {
        return this.rpcClientStatus.get() == RpcClientStatus.SHUTDOWN;
    }

    private void switchServerAsync() {
        switchServerAsync(null, false);
    }

    /**
     * consume by {@link RpcClient#handleReconnectEvent()}
     * @param serverInfo
     * @param onRequestFail
     */
    protected void switchServerAsync(ServerInfo serverInfo,
                                     boolean onRequestFail) {
        reconnectionSignal.offer(new ReconnectContext(serverInfo, onRequestFail));
    }

    protected Response handleServerRequest(final Request request) {
        for (ServerRequestHandler serverRequestHandler : serverRequestHandlers) {
            Response response = serverRequestHandler.requestReply(request, currentConnection);
            if (response == null) {
                continue;
            }

            log.info("request type:{} handle by {}", request.getClass().getSimpleName(), serverRequestHandler.getClass().getSimpleName());
            return response;
        }

        return null;
    }


    public abstract Connection connectToServer(ServerInfo serverInfo);

    public abstract int rpcPortOffset();

    class ConnectResetRequestHandler
            implements ServerRequestHandler {
        @Override
        public Response requestReply(Request request,
                                     Connection connection) {
            if (request instanceof ConnectResetRequest connectResetRequest) {
                synchronized (RpcClient.this) {
                    if (rpcClientStatus.get() == RpcClientStatus.RUNNING) {
                        if (StringUtils.isNotBlank(connectResetRequest.getServerIp())) {
                            ServerInfo serverInfo = new ServerInfo(connectResetRequest.getServerIp(), Integer.parseInt(connectResetRequest.getServerPort()));
                            switchServerAsync(serverInfo, false);
                        } else {
                            switchServerAsync();
                        }
                    }
                }

                return new ConnectResetResponse();
            }

            return null;
        }
    }


    @AllArgsConstructor
    @Getter
    @Setter
    public static class ServerInfo {
        protected String serverIp;
        protected int serverPort;

        public static ServerInfo fromAddress(String address) {
            String[] split = address.split(":");
            return new ServerInfo(split[0], Integer.parseInt(split[1]));
        }

        public String getAddress() {
            return serverIp + ":" + serverPort;
        }
    }

    @AllArgsConstructor
    static class ReconnectContext {
        ServerInfo serverInfo;
        boolean onRequestFail;
    }

    @AllArgsConstructor
    public static class ConnectionEvent {
        public static final int CONNECTED = 1;
        public static final int DISCONNECTED = 0;
        int eventType;
        Connection connection;

        public boolean isConnected() {
            return eventType == CONNECTED;
        }

        public boolean isDisConnected() {
            return eventType == DISCONNECTED;
        }
    }
}
